package com.adam.example.aggregation.pipeline;

import com.adam.example.TransportClientHolder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.filter.Filter;
import org.elasticsearch.search.aggregations.bucket.filter.FilterAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation;
import org.elasticsearch.search.aggregations.metrics.sum.Sum;
import org.elasticsearch.search.aggregations.pipeline.PipelineAggregatorBuilders;
import org.elasticsearch.search.aggregations.pipeline.bucketscript.BucketScriptPipelineAggregationBuilder;

import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;

public class BucketScriptAggregationExample {

    public static void main(String[] args) throws UnknownHostException {
        TransportClient transportClient = TransportClientHolder.getInstance().getTransportClient();

        FilterAggregationBuilder filterAggregationBuilder = AggregationBuilders.filter("t-shirt",
                QueryBuilders.matchQuery("sales_type", "t-shirt"))
                .subAggregation(AggregationBuilders.sum("sales").field("sales_price"));
        Map<String, String> bucketsPathsMap = new HashMap<>();
        bucketsPathsMap.put("tShirtSales", "t-shirt>sales");
        bucketsPathsMap.put("totalSales", "sales");
        BucketScriptPipelineAggregationBuilder bucketScriptPipelineAggregationBuilder = PipelineAggregatorBuilders
                .bucketScript("agg", bucketsPathsMap, new Script("params.tShirtSales / params.totalSales * 100"));
        DateHistogramAggregationBuilder dateHistogramAggregationBuilder = AggregationBuilders
                .dateHistogram("sales_per_month").field("sales_date")
                .dateHistogramInterval(DateHistogramInterval.MONTH)
                .subAggregation(AggregationBuilders.sum("sales").field("sales_price"))
                .subAggregation(filterAggregationBuilder)
                .subAggregation(bucketScriptPipelineAggregationBuilder);

        SearchResponse searchResponse = transportClient.prepareSearch("test")
                .setTypes("sales").setSize(0).addAggregation(dateHistogramAggregationBuilder).get();
        Histogram histogram = searchResponse.getAggregations().get("sales_per_month");
        for(Histogram.Bucket bucket: histogram.getBuckets()) {
            Sum sum1 = bucket.getAggregations().get("sales");
            System.out.println(bucket.getKey() + ":" + sum1.getValue());
            //getKey返回的是org.joda.time.DateTime类型
            Filter filter = bucket.getAggregations().get("t-shirt");
            Sum sum2 = filter.getAggregations().get("sales");
            System.out.println("t-shirt销售额：" + sum2.getValue());
            NumericMetricsAggregation.SingleValue singleValue = bucket.getAggregations().get("agg");
            System.out.println("t-shirt销售额百分比：" + singleValue.value() + "%");
        }

        transportClient.close();
    }

}
